package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.heima.article.config.stream.KafkaStreamListener;
import com.heima.model.article.dtos.ArticleVisitStreamMess;
import com.heima.model.article.dtos.UpdateArticleMess;
import com.heima.model.common.constants.ArticleConstants;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.stereotype.Component;

import java.util.Arrays;

/**
 * @Description:
 * @Version: V1.0
 */
@Component
@Slf4j
public class HotArticleStreamHandler implements KafkaStreamListener<KStream<String, String>> {

    /**
     * 接收到待处理的消息 mess  UpdateArticleMess
     * @return
     */
    @Override
    public String listenerTopic() {
        return ArticleConstants.HOTARTICLE_SCORE_INPUT_TOPIC;
    }

    /**
     * 处理成功后的结果发生给 新topic
     * @return
     */
    @Override
    public String sendTopic() {
        return  ArticleConstants.HOTARTICLE_INCR_HANDLE_OUPUT_TOPIC;
    }

    @Override
    public KStream<String, String> getService(KStream<String, String> stream) {
        return stream.flatMapValues( (key,value)->{

            UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
            log.info("mess is :{} ", mess);
            // 文章id:LIKES  id:VIEWS  id:COLLECT
            Long articleId = mess.getArticleId();
            System.out.println("articleId: "+articleId);
            return Arrays.asList( articleId + "_" + mess.getType());
        })
                .groupBy((key,value) -> value)
                .windowedBy(TimeWindows.of(5000))
                .count()
                .toStream()
                .map((key,value)->{
                    log.info("处理成功后的 k:{}, v:{}",key.key().toString(), value.toString() );
                    return new KeyValue<String,String>(key.key().toString(),
                            formatObj(key.key().toString(), value.toString()));
                });

    }

    /**
     * 处理后的结果 封装到 ArticleVisitStreamMess
     * @param key  id:LIKES
     * @param value 3
     * @return
     */
    private String formatObj(String key, String value) {

        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        String[] split = key.split("_");
        mess.setArticleId(Long.valueOf(split[0]));

        // 阅读
        if (UpdateArticleMess.UpdateArticleType.VIEWS.name().equals(split[1])) {
            mess.setView(Long.valueOf(value));
        }
        if (UpdateArticleMess.UpdateArticleType.LIKES.name().equals(split[1])) {
            mess.setLike(Long.valueOf(value));
        }

        if (UpdateArticleMess.UpdateArticleType.COMMENT.name().equals(split[1])) {
            mess.setComment(Long.valueOf(value));
        }

        if (UpdateArticleMess.UpdateArticleType.COLLECTION.name().equals(split[1])) {
            mess.setCollect(Long.valueOf(value));
        }

        log.info("formatObj mess:{}", mess);
        return JSON.toJSONString(mess);
    }
}
